/****************************************************
 * 创建人:   robin
 * 创建时间: 2023/8/30.030 14:35
 * 项目名称: cpy
 * 文件名称: CollectServerCallable.java
 * 文件描述: 
 *
 * All rights Reserved, Designed By 德品软件
 * @Copyright:2023-2023
 *
 ********************************************************/
package com.dp.threadmanager;

import com.dp.constant.CollectConstant;
import com.dp.enumeration.ReturnCodeEnum;
import com.dp.enumeration.ServerPortEnum;
import com.dp.service.HandleMessageService;
import com.dp.utils.TaskThreadPool;
import com.dp.vo.response.ResponseData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;

import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.Callable;

/**
 * 包名称：com.dp.threadmanager
 * 类名称：CollectServerCallable
 * 类描述：
 * 创建人： robin
 * 创建时间：2023/8/30.030 14:35
 */
@Slf4j
@Configuration
public class CollectServerCallable implements Callable<Void> {
    
    private HandleMessageService handleMessageService;
    private ServerPortEnum serverPortEnum;
    private Selector selector = null;
    private ServerSocketChannel serverSocketChannel = null;
//    private ByteBuffer recvByteBuffer = ByteBuffer.allocate(CollectConstant.MSG_BUF_LEN).order(ByteOrder.BIG_ENDIAN);
    
    public CollectServerCallable(HandleMessageService handleMessageService) {
        this.handleMessageService = handleMessageService;
    }
    
    public void setServerPortEnum(ServerPortEnum serverPortEnum) {
        this.serverPortEnum = serverPortEnum;
    }
    
    @Override
    public Void call() throws Exception {
        
        log.info("CollectServerCallable start");
        
        try {
            while (null == serverPortEnum) {
                Thread.sleep(CollectConstant.SECOND_1); // 休眠1秒
            }
            
            if (0 == start()) {
                process();
            }
        } catch (Exception e) {
            log.error("CollectServerCallable failed:{}", e.getMessage());
        } finally {
            stop();
        }
        log.info("CollectServerCallable end");
        return null;
    }
    
    private int start(){
        int returnCode = 0;
        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            log.info("CollectServerCallable at port: {}", serverPortEnum.getCode());
            serverSocketChannel.socket().bind(new InetSocketAddress(serverPortEnum.getCode())); // 绑定监听端口
            serverSocketChannel.socket().setReuseAddress(true); // 设置 SO_REUSEADDR 选项
            serverSocketChannel.configureBlocking(false);   // 设置非阻塞模式
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, serverSocketChannel);    // 注册 accept 事件
            
            log.info("CollectServerCallable at port: {}", serverPortEnum.getCode());
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectServerCallable [{}] start failed:{}", serverPortEnum.getCode(), e.getMessage());
            returnCode = -1;
        }
        return returnCode;
    }
    
    private void process() {
        try {
            while (true) {
                if (selector.select(CollectConstant.SECOND_1) == 0) {
                    log.debug("CollectServerCallable[{}] no events", serverPortEnum.getCode());
                    continue;
                }
                
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
                
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    if (key != null) {
                        if (key.isAcceptable()) {
                            log.debug("CollectServerCallable[{}]:a connection was accepted by a ServerSocketChannel", serverPortEnum.getCode());
                            SocketChannel socketChannel = ((ServerSocketChannel) key.attachment()).accept();
    
                            CollectCommCallable collectCommCallable = new CollectCommCallable(handleMessageService);
                            collectCommCallable.setSocketChannel(socketChannel);
                            collectCommCallable.setServerPortEnum(serverPortEnum);
                            TaskThreadPool.add(collectCommCallable);
                        } else if (key.isConnectable()) {
                            log.debug("CollectServerCallable[{}]:a connection was established with a remote server", serverPortEnum.getCode());
                        } else if (key.isReadable()) {
                            log.debug("CollectServerCallable[{}]:a channel is ready for reading", serverPortEnum.getCode());
                        } else if (key.isWritable()) {
                            log.debug("CollectServerCallable[{}]:a channel is ready for writing", serverPortEnum.getCode());
                        } else {
                            log.debug("CollectServerCallable[{}]:unknow SelectionKey", serverPortEnum.getCode());
                        }
                    }
                    keyIterator.remove();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectServerCallable[{}] process failed:{}", serverPortEnum.getCode(), e.getMessage());
        }
    }
    
    private void stop(){
        try {
            if (serverSocketChannel != null){
                serverSocketChannel.close();
                serverSocketChannel = null;
            }
            if (selector != null) {
                selector.close();
                selector = null;
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("CollectServerCallable[{}] stop failed:{}", serverPortEnum != null ? serverPortEnum.getCode() : 0, e.getMessage());
        }
    }
    
}
